package org.apache.kafka.niezhipeng.customSerializerDemo;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.niezhipeng.customSerializerDemo.pojo.Company;

import java.util.Properties;

/**
 * @author Nie ZhiPeng
 * @date 2025/7/17 23:20
 */
public class CompanyProducer {



    public static void main(String[] args) {
        Properties properties = CompanyDemoUtil.getProducerProperties();

        KafkaProducer<String, Company> kafkaProducer = new KafkaProducer<>(properties);
        sendAndForgot(kafkaProducer, new Company("公司1", "地址1"));
        sendAndForgot(kafkaProducer, new Company("公司2", "地址2"));
        sendAndForgot(kafkaProducer, null);
        // close方法会等待所有消息发送完成后再关闭
        kafkaProducer.close();
    }

    /**
     * 发送消息，不等待消息确认成功
     *
     * @param kafkaProducer
     * @param message
     */
    private static void sendAndForgot(KafkaProducer<String, Company> kafkaProducer, Company message) {
        ProducerRecord<String, Company> record = new ProducerRecord<>(CompanyDemoUtil.TOPIC, message);
        try {
            kafkaProducer.send(record);
        } catch (Exception e) {
            System.err.println("向Kafka发送消息失败");
            e.printStackTrace();
        }
    }

    /**
     * 同步发送消息
     *
     * @param kafkaProducer
     * @param message
     */
    private static void sendSync(KafkaProducer<String, Company> kafkaProducer, Company message) {
        ProducerRecord<String, Company> record = new ProducerRecord<>(CompanyDemoUtil.TOPIC, message);
        try {
            RecordMetadata recordMetadata = kafkaProducer.send(record).get();
        } catch (Exception e) {
            System.err.println("向Kafka发送消息失败");
            e.printStackTrace();
        }
    }

    /**
     * 同步发送消息
     *
     * @param kafkaProducer
     * @param message
     */
    private static void sendAndCallBack(KafkaProducer<String, Company> kafkaProducer, Company message) {
        ProducerRecord<String, Company> record = new ProducerRecord<>(CompanyDemoUtil.TOPIC, message);
        try {
            kafkaProducer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        exception.printStackTrace();
                    } else {
                        System.out.println("发送成功");
                    }
                }
            });
        } catch (Exception e) {
            System.err.println("向Kafka发送消息失败");
            e.printStackTrace();
        }
    }
}
